package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;

public class Demo14_UserDefinedAggregateFunction {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        tenv.executeSql("create temporary view stu_info as  " +
                "select 1 as id, 'zs' as name , 'male' as gender, 40.0 as score " +
                "union all " +
                "select 2 , 'hh' , 'male', 60.0 " +
                "union all " +
                "select 3 , 'xx' , 'female', 100.0 " +
                "union all " +
                "select 4 , 'yy' , 'female', 120.0 "
        );

        /**
         * +----+-----+--------+-----------+--------+
         * | op |  id |   name |    gender |  score |
         * +----+-----+--------+-----------+--------+
         * | +I |   1 |     zs |      male |   40.0 |
         * | +I |   2 |     hh |      male |   60.0 |
         * | +I |   3 |     xx |    female |  100.0 |
         * | +I |   4 |     yy |    female |  120.0 |
         * +----+-----+--------+-----------+--------+
         */
        //tenv.executeSql("select * from stu_info").print();

        tenv.createTemporaryFunction("my_avg", MyAvgFunction.class);
        tenv.executeSql("select  gender,my_avg(score)  from stu_info group by gender").print();


    }


    public static class MyAcc {
        public int cnt;
        public double sum;
    }


    /**
     * 自定义聚合函数：用于求平均值
     */
    public static class MyAvgFunction extends AggregateFunction<Double,MyAcc>{

        /**
         *  返回最终结果
         */
        @Override
        public Double getValue(MyAcc accumulator) {
            return accumulator.sum/ accumulator.cnt;
        }

        /**
         * 创建累加器
         */
        @Override
        public MyAcc createAccumulator() {
            return new MyAcc();
        }


        /**
         * 单步聚合：来一条数据，聚合到累加器
         */
        public void accumulate(MyAcc acc, Double score){
            acc.cnt += 1;
            acc.sum += score;
        }


        /**
         * 数据发生撤回，累加器该如何修正
         */
        public void retract(MyAcc acc, Double score){

            acc.cnt -= 1;
            acc.sum -= score;

        }


        /**
         * 合并多个累加器
         */
        public void merge(MyAcc accumulator, Iterable<MyAcc> it) {

            for (MyAcc acc : it) {
                accumulator.cnt += acc.cnt;
                accumulator.sum += acc.sum;
            }

        }


        /**
         * 累加器重置
         */
        public void resetAccumulator(MyAcc acc) {
            acc.cnt = 0;
            acc.sum = 0;
        }

    }

}
